--- title: Testing Diffusion Flow Embedder on Circle and Swiss Roll keywords: fastai sidebar: home_sidebar nb_path: "02d02a Testing Diffusion Flow Embedder on Circle and Swiss Roll.ipynb" ---
from directed_graphs.datasets import directed_circle, directed_cylinder, directed_spiral, directed_swiss_roll, directed_spiral_uniform, directed_swiss_roll_uniform
from directed_graphs.datasets import plot_directed_2d, plot_directed_3d
from directed_graphs.diffusion_flow_embedding import DiffusionFlowEmbedder
import torch
import numpy as np
if torch.__version__[:4] == "1.13":
# device = torch.device('cuda' if torch.cuda.is_available() else 'mps' if torch.has_mps else 'cpu')
device = torch.device('cuda' if torch.cuda.is_available() else 'cpu')
else:
device = torch.device('cuda' if torch.cuda.is_available() else 'cpu')
device
X, flow, labels = directed_circle(num_nodes=300, radius=1)
plot_directed_2d(X, flow, labels)
X = torch.tensor(X)
flow = torch.tensor(flow)
X = X.float().to(device)
flow = flow.float().to(device)
dfe = DiffusionFlowEmbedder(X,
flow,
t=1,
sigma_graph=15,
sigma_embedding=15,
device=device,
autoencoder_shape = [50,10],
flow_artist_shape=[30,20,10],
flow_strength_graph=2,
flow_strength_embedding=5,
learnable_flow_strength=True,
weight_of_flow = 0.5,
learning_rate=0.001
)
dfe = dfe.to(device)
embeddings = dfe.fit(n_steps=2000)
dfe.visualize_points(labels)
X, flow, labels = directed_circle(num_nodes=1000, radius=1, ytilt=np.pi/4)
plot_directed_3d(X, flow, labels, mask_prob=0.5)
X = torch.tensor(X)
flow = torch.tensor(flow)
X = X.float().to(device)
flow = flow.float().to(device)
dfe = DiffusionFlowEmbedder(X,
flow,
t=1,
sigma_graph=15,
sigma_embedding=15,
device=device,
autoencoder_shape = [20,10],
flow_artist_shape=[30,20,10],
flow_strength_graph=2,
flow_strength_embedding=5,
learnable_flow_strength=True,
weight_of_flow = 0.5,
learning_rate=0.01
)
dfe = dfe.to(device)
embeddings = dfe.fit(n_steps=2000)
dfe.visualize_points(labels)
X, flow, labels = directed_cylinder(num_nodes=1000, radius=1, height=20, xtilt=np.pi/4, ytilt=0)
plot_directed_3d(X, flow, labels, mask_prob=0.5)
X = torch.tensor(X)
flow = torch.tensor(flow)
X = X.float().to(device)
flow = flow.float().to(device)
dfe = DiffusionFlowEmbedder(X,flow,t=1,sigma_graph=1,sigma_embedding=1,flow_strength_graph=2
device=device)
dfe = dfe.to(device)
embeddings = dfe.fit(n_steps=2000)
dfe.visualize_points(labels)
X, flow, labels = directed_spiral(num_nodes=1000, num_spirals=2.5, radius=1)
plot_directed_2d(X, flow, labels)
X = torch.tensor(X)
flow = torch.tensor(flow)
X = X.float().to(device)
flow = flow.float().to(device)
dfe = DiffusionFlowEmbedder(X,flow,t=1,sigma_graph=10,sigma_embedding=15,
device=device,
learning_rate=0.001,
weight_of_flow=1)
dfe = dfe.to(device)
embeddings = dfe.fit(n_steps=10000)
dfe.visualize_points(labels)
X, flow, labels = directed_spiral(num_nodes=100, num_spirals=2.5, radius=1, xtilt=np.pi/4)
plot_directed_3d(X, flow, labels, mask_prob=0.5)
X = torch.tensor(X)
flow = torch.tensor(flow)
X = X.float().to(device)
flow = flow.float().to(device)
dfe = DiffusionFlowEmbedder(X,flow,t=1,sigma_graph=15,sigma_embedding=15,device=device)
dfe = dfe.to(device)
embeddings = dfe.fit(n_steps=2000)
dfe.visualize_points(labels)
X, flow, labels = directed_swiss_roll(num_nodes=1000, num_spirals=2.5, radius=1, height=20, xtilt=np.pi/4, ytilt=0)
plot_directed_3d(X, flow, labels, mask_prob=0.5)
X = torch.tensor(X)
flow = torch.tensor(flow)
X = X.float().to(device)
flow = flow.float().to(device)
dfe = DiffusionFlowEmbedder(X,flow,t=1,sigma_graph=15,sigma_embedding=15,device=device)
dfe = dfe.to(device)
embeddings = dfe.fit(n_steps=2000)
dfe.visualize_points(labels)
import phate
phate_operator = phate.PHATE(n_components = 2, knn=5, decay=20, t=150)
phate_X = phate_operator.fit_transform(X.cpu())
phate.plot.scatter2d(phate_X, c=labels)
X, flow, labels = directed_swiss_roll_uniform(num_nodes=1000, num_spirals=2.5, radius=1, height=20, xtilt=0, ytilt=0)
plot_directed_3d(X, flow, labels, mask_prob=0.5)
X = torch.tensor(X)
flow = torch.tensor(flow)
X = X.float().to(device)
flow = flow.float().to(device)
dfe = DiffusionFlowEmbedder(X,flow,t=1,sigma_graph=15,sigma_embedding=15,device=device)
dfe = dfe.to(device)
embeddings = dfe.fit(n_steps=2000)
dfe.visualize_points(labels)